Conversation
Merge to main
chore : add gitignore
Walkthrough음성 케어 기능을 추가하고 감정 분석 파이프라인을 개선했습니다. VoiceAnalyze에 surprise_bps 필드를 추가하고, 음성 업로드 시 비동기 감정 분석을 수행하며, 케어기버 대시 보드용 API 엔드포인트를 새로 구현했습니다. Changes
Sequence Diagram(s)sequenceDiagram
actor User as 사용자
participant API as FastAPI
participant VoiceService as VoiceService
participant STTTask as STT 백그라운드<br/>작업
participant EmotionTask as 감정 분석<br/>백그라운드 작업
participant DB as DatabaseService
User->>API: POST /users/voices<br/>(오디오 파일)
API->>VoiceService: upload_user_voice()
rect rgb(200, 220, 255)
Note over VoiceService,DB: 즉시 응답 반환
VoiceService->>DB: 음성 메타데이터 저장
VoiceService-->>API: 응답 반환 (voice_id)
end
API-->>User: 202 응답 (voice_id)
rect rgb(255, 220, 200)
Note over STTTask,DB: 병렬 비동기 처리
par STT 및 NLP 분석
STTTask->>STTTask: 음성 → 텍스트 변환
STTTask->>STTTask: 텍스트 감정 분석
STTTask->>DB: voice_content 저장
and 오디오 감정 분석
EmotionTask->>EmotionTask: 오디오 로드 및 처리
EmotionTask->>EmotionTask: 감정 점수 추출
EmotionTask->>EmotionTask: 점수 정규화<br/>(0-10000)
EmotionTask->>DB: VoiceAnalyze 저장<br/>(happy/sad/neutral/angry/fear/surprise)
end
end
User->>API: GET /users/voices/{voice_id}
API->>VoiceService: get_user_voice_detail()
VoiceService->>DB: 분석 결과 조회
DB-->>VoiceService: 음성 및 감정 데이터
VoiceService-->>API: 응답 (분석 결과 포함)
API-->>User: 분석 완료 데이터
sequenceDiagram
actor Caregiver as 케어기버
participant API as FastAPI
participant CareService as CareService
participant VoiceService as VoiceService
participant DB as DatabaseService
Caregiver->>API: GET /care/users/voices<br/>(care_username)
API->>CareService: 초기화
API->>VoiceService: get_care_voice_list()
rect rgb(220, 240, 220)
Note over VoiceService,DB: 케어 대상자의 음성 조회
VoiceService->>DB: get_care_voices()
DB-->>VoiceService: 음성 목록<br/>(생성일, 감정)
end
VoiceService-->>API: 응답 (CareUserVoiceListResponse)
API-->>Caregiver: 음성 목록
Caregiver->>API: GET /care/users/voices/analyzing/frequency<br/>(care_username, month)
API->>CareService: get_emotion_monthly_frequency()
rect rgb(255, 240, 200)
Note over CareService,DB: 월간 감정 빈도 집계
CareService->>DB: 월간 음성 조회
DB-->>CareService: 감정 데이터
CareService->>CareService: 감정별 빈도 계산
end
CareService-->>API: 응답 (감정 빈도 맵)
API-->>Caregiver: 월간 감정 통계
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes 추가 검토 필요 영역:
Possibly related PRs
Poem
Pre-merge checks and finishing touches❌ Failed checks (3 warnings)
✨ Finishing touches
🧪 Generate unit tests (beta)
Warning Review ran into problems🔥 ProblemsGit: Failed to clone repository. Please run the Comment |
There was a problem hiding this comment.
Actionable comments posted: 7
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
app/main.py (1)
169-188: username 매개변수를 필수로 지정하세요.Line 173에서
username: str = None으로 선택적 매개변수로 정의했지만, Line 177-178에서 필수로 체크하고 있습니다. API 계약을 명확히 하기 위해 처음부터 필수로 지정하는 것을 권장합니다.@users_router.post("/voices", response_model=VoiceQuestionUploadResponse) async def upload_voice_with_question( file: UploadFile = File(...), question_id: int = Form(...), - username: str = None, + username: str = Form(...), ): db = next(get_db()) voice_service = get_voice_service(db) - if not username: - raise HTTPException(status_code=400, detail="username is required as query parameter") result = await voice_service.upload_voice_with_question(file, username, question_id)
🧹 Nitpick comments (6)
restart_dev.sh (1)
7-8: 하드코딩된 경로를 환경 변수로 변경하세요.
/home/ubuntu/caring-voice경로가 하드코딩되어 있어 다른 환경에서 작동하지 않습니다.다음과 같이 개선하세요:
-PROJECT_DIR="/home/ubuntu/caring-voice" +PROJECT_DIR="${PROJECT_DIR:-$(pwd)}" cd $PROJECT_DIRapp/db_service.py (1)
92-113: 연결 사용자 조회 로직을 단일 쿼리로 최적화할 수 있습니다.현재 구현은 보호자 조회 → 연결 사용자 조회 → 음성 조회로 3단계 쿼리를 실행합니다. 성능 향상을 위해 조인을 활용한 단일 쿼리로 리팩토링을 고려해보세요.
예시:
def get_care_voices(self, care_username: str, skip: int = 0, limit: int = 20) -> List[Voice]: """보호자(care)의 연결 사용자 음성 중 voice_analyze가 존재하는 항목만 최신순 조회""" from sqlalchemy.orm import joinedload, aliased CareUser = aliased(User) LinkedUser = aliased(User) q = ( self.db.query(Voice) .join(LinkedUser, Voice.user_id == LinkedUser.user_id) .join(CareUser, CareUser.connecting_user_code == LinkedUser.user_code) .join(VoiceAnalyze, VoiceAnalyze.voice_id == Voice.voice_id) .filter(CareUser.username == care_username) .options(joinedload(Voice.questions), joinedload(Voice.voice_analyze)) .order_by(Voice.created_at.desc()) .offset(skip) .limit(limit) ) return q.all()app/voice_service.py (3)
81-83: 백그라운드 태스크의 실패를 추적할 수 없습니다.
asyncio.create_task로 생성된 태스크는 fire-and-forget 방식이므로 실패 시 호출자가 알 수 없습니다. 프로덕션 환경에서는 태스크 완료 상태와 에러를 추적할 수 있는 메커니즘(예: 로깅, 메트릭, 태스크 큐)을 고려하세요.
313-328: 예외 처리가 너무 광범위합니다.모든 예외를 무시하고 빈 리스트를 반환하면 문제를 파악하기 어렵습니다. 최소한 로깅을 추가하거나 예외를 상위로 전파하는 것을 권장합니다.
def get_care_voice_list(self, care_username: str, skip: int = 0, limit: int = 20) -> Dict[str, Any]: """보호자 페이지: 연결된 사용자의 분석 완료 음성 목록 조회(페이징)""" try: voices = self.db_service.get_care_voices(care_username, skip=skip, limit=limit) items = [] for v in voices: created_at = v.created_at.isoformat() if v.created_at else "" emotion = v.voice_analyze.top_emotion if v.voice_analyze else None items.append({ "voice_id": v.voice_id, "created_at": created_at, "emotion": emotion, }) return {"success": True, "voices": items} - except Exception: + except Exception as e: + print(f"get_care_voice_list error for {care_username}: {e}", flush=True) return {"success": False, "voices": []}
330-359: 예외 처리에 로깅을 추가하세요.
get_care_voice_list와 동일하게, 예외 발생 시 로깅을 추가하여 문제 추적을 용이하게 하세요.def get_user_voice_detail(self, voice_id: int, username: str) -> Dict[str, Any]: """voice_id와 username으로 상세 정보 조회""" try: voice = self.db_service.get_voice_detail_for_username(voice_id, username) if not voice: return {"success": False, "error": "Voice not found or not owned by user"} # ... rest of the method - except Exception: + except Exception as e: + print(f"get_user_voice_detail error for voice_id={voice_id}, username={username}: {e}", flush=True) return {"success": False, "error": "Failed to fetch voice detail"}app/main.py (1)
202-214: random 모듈이 중복 import되고 있습니다.Line 27에서 이미
import random을 했는데 Line 208에서 다시 import하고 있습니다. 모듈 레벨 import를 사용하세요.@questions_router.get("/random") async def get_random_question(): db = next(get_db()) question_count = db.query(Question).count() if question_count == 0: return {"success": False, "question": None} - import random offset = random.randint(0, question_count - 1)또한, 질문 테이블이 커질 경우 COUNT 쿼리의 성능을 고려하여 캐싱을 검토하세요.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (12)
.gitignore(0 hunks)README.md(1 hunks)_schema_fix.py(1 hunks)app/care_service.py(1 hunks)app/db_service.py(8 hunks)app/dto.py(2 hunks)app/emotion_service.py(4 hunks)app/main.py(4 hunks)app/models.py(2 hunks)app/stt_service.py(2 hunks)app/voice_service.py(6 hunks)restart_dev.sh(1 hunks)
💤 Files with no reviewable changes (1)
- .gitignore
🔇 Additional comments (22)
README.md (1)
32-40: 비동기 처리 흐름 문서화가 명확합니다.업로드 후 비동기 처리 흐름에 대한 설명이 잘 작성되었습니다. 사용자가 API 동작 방식을 이해하는 데 도움이 됩니다.
app/models.py (2)
94-94: surprise_bps 필드 추가가 적절합니다.새로운 감정 카테고리를 올바르게 추가했으며, 기본값 0과 NOT NULL 제약 조건이 적절합니다.
106-108: 데이터 무결성 제약 조건이 올바르게 업데이트되었습니다.surprise_bps를 범위 체크와 합계 체크에 모두 포함시켜 6개 감정의 총합이 10000이 되도록 보장합니다.
app/emotion_service.py (2)
20-22: rebalanced 모델로 변경이 적절합니다.데이터 불균형을 해결한 모델로 교체한 것은 감정 분석 정확도 향상에 도움이 됩니다.
166-183: 한글→영문 라벨 매핑이 유용합니다.모델이 한글 라벨을 반환할 수 있는 경우를 대비한 매핑이 잘 구현되었습니다.
app/dto.py (2)
66-66: voice_id 필드 추가가 적절합니다.API 응답에 voice_id를 포함시켜 클라이언트가 음성을 식별할 수 있게 했습니다.
78-108: 새로운 Care 관련 DTO가 잘 정의되었습니다.보호자 기능을 위한 DTO들이 명확하게 정의되었으며,
VoiceAnalyzePreviewResponse에surprise_bps필드가 모델 변경과 일관되게 포함되었습니다.app/care_service.py (3)
12-44: 월별 감정 빈도 집계 로직이 올바릅니다.보호자-사용자 연결 검증, 월 파싱, 데이터베이스 쿼리가 적절하게 구현되었습니다. 에러 처리도 잘 되어 있습니다.
65-70: 주차 계산 로직을 검증하세요.주차 계산이
(week-1)*7+1부터week*7까지로 되어 있어, 항상 일요일 시작 기준이 아닙니다. 예를 들어:
- week=1: day 1~7
- week=2: day 8~14
이 방식은 달력 주차(월요일 시작)와 다를 수 있습니다. 비즈니스 요구사항에 맞는지 확인하세요.
주차 계산 방식이 요구사항과 일치하는지 확인이 필요합니다. ISO 주차 또는 다른 기준이 필요한 경우
datetime.isocalendar()를 고려하세요:# ISO week (Monday-Sunday) 기준 예시 from datetime import datetime dt = datetime(y, m, 1) iso_year, iso_week, iso_weekday = dt.isocalendar()
93-95: 동점 처리 로직이 합리적입니다.여러 감정이 동일한 빈도를 가질 때 가장 먼저 업로드된 감정을 선택하는 방식이 일관성 있고 결정론적입니다.
restart_dev.sh (1)
39-39: 헬스체크 엔드포인트 존재 여부를 확인하세요.스크립트가
/health엔드포인트를 호출하지만, 제공된 파일에서 이 엔드포인트의 구현을 찾을 수 없습니다. app/main.py에 헬스체크 엔드포인트가 구현되어 있는지 확인하세요.다음 스크립트로 헬스체크 엔드포인트 구현을 확인하세요:
app/db_service.py (5)
40-42: 구현이 올바릅니다.기존 패턴을 따르는 간단한 조회 메서드입니다.
66-79: 소유권 검증과 eager loading이 적절합니다.username을 통한 소유권 검증과 joinedload를 활용한 관련 엔티티 로딩이 올바르게 구현되어 있습니다.
170-190: surprise_bps 필드 추가가 적절합니다.기존 감정 필드들과 일관된 패턴으로 구현되어 있습니다.
196-219: 선택적 업데이트 로직이 올바릅니다.다른 필드들과 동일한 패턴으로 surprise_bps의 선택적 업데이트가 구현되어 있습니다.
278-285: 소유권 검증 메서드가 적절합니다.삭제 전 소유권 확인 용도로 적합한 구현입니다.
app/voice_service.py (3)
176-182: 다양한 키 이름을 처리하는 로직이 적절합니다.감정 서비스가 반환할 수 있는 다양한 키 이름(happy/happiness, sad/sadness 등)에 대한 fallback 처리가 잘 되어 있습니다.
235-247: DB 저장 로직이 적절합니다.모든 감정 필드와 surprise_bps를 포함하여 올바르게 저장하고 있습니다.
192-223: 정규화 로직이 올바르게 구현되어 있습니다.감정 점수를 10000으로 정규화하고 반올림 오차를 가장 큰 값에 보정하는 로직이 적절합니다. 로깅도 충분하여 디버깅이 용이합니다.
엣지 케이스 검증을 위해 다음 스크립트를 실행하여 정규화 로직을 테스트할 수 있습니다:
app/main.py (3)
31-36: 라우터 기반 모듈화가 잘 구현되어 있습니다.도메인별로 라우터를 분리하고 적절한 prefix와 태그를 사용하여 API 구조가 명확합니다.
Also applies to: 339-344
217-244: care_username 기반 접근 제어를 검증하세요.care 엔드포인트들이 care_username 쿼리 파라미터에만 의존하여 사용자를 식별합니다. 인증된 사용자의 세션/토큰과 care_username을 검증하는 로직이 없다면, 임의의 사용자가 다른 보호자의 데이터에 접근할 수 있는 보안 취약점이 됩니다.
다음 스크립트로 인증 미들웨어나 의존성 주입이 있는지 확인하세요:
118-135: role 파라미터 검증을 확인하세요.Line 119에서 role이 쿼리 파라미터로 전달되고 Line 125에서 auth_service.signin에 전달됩니다. signin 메서드가 사용자의 실제 role과 요청된 role을 검증하는지 확인하세요. 그렇지 않으면 권한 상승 취약점이 될 수 있습니다.
다음 스크립트로 signin 메서드의 role 검증 로직을 확인하세요:
| from app.database import engine | ||
| from sqlalchemy import text | ||
|
|
||
|
|
||
| def ensure_surprise_column() -> None: | ||
| with engine.connect() as conn: | ||
| res = conn.execute(text("SHOW COLUMNS FROM voice_analyze LIKE 'surprise_bps'")) | ||
| row = res.fetchone() | ||
| if row is None: | ||
| print("Adding surprise_bps column...") | ||
| conn.execute(text("ALTER TABLE voice_analyze ADD COLUMN surprise_bps SMALLINT NOT NULL DEFAULT 0")) | ||
| print("Added surprise_bps") | ||
| else: | ||
| print("surprise_bps exists") | ||
| conn.commit() | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| ensure_surprise_column() | ||
|
|
||
|
|
||
|
|
There was a problem hiding this comment.
Alembic 마이그레이션 사용을 권장합니다.
이 스크립트는 수동으로 스키마를 변경하지만, README.md에 Alembic 마이그레이션 가이드가 이미 있습니다. 몇 가지 문제점:
- 원시 SQL 사용으로 마이그레이션 히스토리 추적 불가
- MySQL 전용 구문(
SHOW COLUMNS)으로 다른 DB와 호환 불가 - 롤백 메커니즘 없음
- 프로덕션 환경에서 실수로 실행될 위험
대신 Alembic을 사용하세요:
alembic revision --autogenerate -m "Add surprise_bps to VoiceAnalyze"
alembic upgrade head🤖 Prompt for AI Agents
_schema_fix.py lines 1-22: this ad-hoc script mutates schema with raw,
MySQL-specific SQL, lacks migration history, downgrade support and safety for
prod; replace it with an Alembic migration: create a new revision (alembic
revision --autogenerate -m "Add surprise_bps to voice_analyze"), implement the
upgrade() to add the surprise_bps SMALLINT with a default and nullable/NOT NULL
handling compatible with our DB, implement downgrade() to drop the column, run
alembic upgrade head to apply, remove or disable this standalone script from
repo/CI so it cannot be run against production, and include a brief test or
checklist to verify the migration applies cleanly across environments.
| def delete_voice_with_relations(self, voice_id: int) -> bool: | ||
| """연관 데이터(voice_question, voice_content, voice_analyze) 삭제 후 voice 삭제""" | ||
| # voice_question | ||
| self.db.query(VoiceQuestion).filter(VoiceQuestion.voice_id == voice_id).delete(synchronize_session=False) | ||
| # voice_content | ||
| self.db.query(VoiceContent).filter(VoiceContent.voice_id == voice_id).delete(synchronize_session=False) | ||
| # voice_analyze | ||
| self.db.query(VoiceAnalyze).filter(VoiceAnalyze.voice_id == voice_id).delete(synchronize_session=False) | ||
| # voice | ||
| deleted = self.db.query(Voice).filter(Voice.voice_id == voice_id).delete(synchronize_session=False) | ||
| self.db.commit() | ||
| return deleted > 0 |
There was a problem hiding this comment.
부분 삭제 가능성에 대한 트랜잭션 처리를 검토하세요.
현재 구현은 여러 테이블에서 순차적으로 삭제를 수행하지만, 중간에 실패 시 부분 삭제가 발생할 수 있습니다. 명시적인 트랜잭션 컨텍스트를 사용하거나 예외 발생 시 롤백 처리를 추가하는 것을 권장합니다.
예시:
def delete_voice_with_relations(self, voice_id: int) -> bool:
"""연관 데이터(voice_question, voice_content, voice_analyze) 삭제 후 voice 삭제"""
- # voice_question
- self.db.query(VoiceQuestion).filter(VoiceQuestion.voice_id == voice_id).delete(synchronize_session=False)
- # voice_content
- self.db.query(VoiceContent).filter(VoiceContent.voice_id == voice_id).delete(synchronize_session=False)
- # voice_analyze
- self.db.query(VoiceAnalyze).filter(VoiceAnalyze.voice_id == voice_id).delete(synchronize_session=False)
- # voice
- deleted = self.db.query(Voice).filter(Voice.voice_id == voice_id).delete(synchronize_session=False)
- self.db.commit()
- return deleted > 0
+ try:
+ # voice_question
+ self.db.query(VoiceQuestion).filter(VoiceQuestion.voice_id == voice_id).delete(synchronize_session=False)
+ # voice_content
+ self.db.query(VoiceContent).filter(VoiceContent.voice_id == voice_id).delete(synchronize_session=False)
+ # voice_analyze
+ self.db.query(VoiceAnalyze).filter(VoiceAnalyze.voice_id == voice_id).delete(synchronize_session=False)
+ # voice
+ deleted = self.db.query(Voice).filter(Voice.voice_id == voice_id).delete(synchronize_session=False)
+ self.db.commit()
+ return deleted > 0
+ except Exception as e:
+ self.db.rollback()
+ raise📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def delete_voice_with_relations(self, voice_id: int) -> bool: | |
| """연관 데이터(voice_question, voice_content, voice_analyze) 삭제 후 voice 삭제""" | |
| # voice_question | |
| self.db.query(VoiceQuestion).filter(VoiceQuestion.voice_id == voice_id).delete(synchronize_session=False) | |
| # voice_content | |
| self.db.query(VoiceContent).filter(VoiceContent.voice_id == voice_id).delete(synchronize_session=False) | |
| # voice_analyze | |
| self.db.query(VoiceAnalyze).filter(VoiceAnalyze.voice_id == voice_id).delete(synchronize_session=False) | |
| # voice | |
| deleted = self.db.query(Voice).filter(Voice.voice_id == voice_id).delete(synchronize_session=False) | |
| self.db.commit() | |
| return deleted > 0 | |
| def delete_voice_with_relations(self, voice_id: int) -> bool: | |
| """연관 데이터(voice_question, voice_content, voice_analyze) 삭제 후 voice 삭제""" | |
| try: | |
| # voice_question | |
| self.db.query(VoiceQuestion).filter(VoiceQuestion.voice_id == voice_id).delete(synchronize_session=False) | |
| # voice_content | |
| self.db.query(VoiceContent).filter(VoiceContent.voice_id == voice_id).delete(synchronize_session=False) | |
| # voice_analyze | |
| self.db.query(VoiceAnalyze).filter(VoiceAnalyze.voice_id == voice_id).delete(synchronize_session=False) | |
| # voice | |
| deleted = self.db.query(Voice).filter(Voice.voice_id == voice_id).delete(synchronize_session=False) | |
| self.db.commit() | |
| return deleted > 0 | |
| except Exception as e: | |
| self.db.rollback() | |
| raise |
🤖 Prompt for AI Agents
In app/db_service.py around lines 287 to 298, the sequential deletes across
multiple tables are not wrapped in an explicit transaction so a failure mid-way
can leave partial deletes; wrap the entire operation in a DB transaction (or
session.begin()) and handle exceptions by rolling back on error, or use the
ORM/session context manager to ensure atomicity and commit only on success, then
return the deletion outcome.
| try: | ||
| print(f"[emotion] start analyze filename={getattr(audio_file,'filename',None)}", flush=True) | ||
| except Exception: | ||
| pass | ||
| # 업로드 확장자 반영하여 임시 파일로 저장 | ||
| import os | ||
| orig_name = getattr(audio_file, "filename", "") or "" | ||
| _, ext = os.path.splitext(orig_name) | ||
| suffix = ext if ext.lower() in [".wav", ".m4a", ".mp3", ".flac", ".ogg", ".aac", ".caf"] else ".wav" | ||
| with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp_file: | ||
| content = audio_file.file.read() | ||
| audio_file.file.seek(0) | ||
| tmp_file.write(content) | ||
| tmp_file_path = tmp_file.name | ||
| try: | ||
| import os as _os | ||
| sz = _os.path.getsize(tmp_file_path) | ||
| print(f"[emotion] tmp saved path={tmp_file_path} size={sz}", flush=True) | ||
| except Exception: | ||
| pass | ||
|
|
||
| # 오디오 로드 (16kHz로 리샘플링) | ||
| audio, sr = librosa.load(tmp_file_path, sr=16000) | ||
| # 오디오 로드 (16kHz, 견고한 로더) | ||
| def robust_load(path: str, target_sr: int = 16000): | ||
| try: | ||
| data, sr = sf.read(path, always_2d=True, dtype="float32") | ||
| if data.ndim == 2 and data.shape[1] > 1: | ||
| data = data.mean(axis=1) | ||
| else: | ||
| data = data.reshape(-1) | ||
| if sr != target_sr: | ||
| data = librosa.resample(data, orig_sr=sr, target_sr=target_sr) | ||
| sr = target_sr | ||
| try: | ||
| print(f"[emotion] robust_load: backend=sf sr={sr} len={len(data)} min={float(np.min(data)):.4f} max={float(np.max(data)):.4f}", flush=True) | ||
| except Exception: | ||
| pass | ||
| return data, sr | ||
| except Exception: | ||
| y, sr = librosa.load(path, sr=target_sr, mono=True) | ||
| y = y.astype("float32") | ||
| try: | ||
| print(f"[emotion] robust_load: backend=librosa sr={sr} len={len(y)} min={float(np.min(y)):.4f} max={float(np.max(y)):.4f}", flush=True) | ||
| except Exception: | ||
| pass | ||
| return y, sr |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
코드 중복 및 로깅 개선이 필요합니다.
여러 가지 문제점:
robust_load함수가app/stt_service.py와 중복됨 (이미 stt_service 리뷰에서 언급)- Line 57에서
import os가 중복 (이미 상단에 import됨) print()대신 Pythonlogging모듈 사용 권장
중복 import 제거:
try:
- # 업로드 확장자 반영하여 임시 파일로 저장
- import os
orig_name = getattr(audio_file, "filename", "") or ""로깅 개선을 위해 파일 상단에 추가:
import logging
logger = logging.getLogger(__name__)그리고 print 문을 교체:
- print(f"[emotion] start analyze filename={getattr(audio_file,'filename',None)}", flush=True)
+ logger.info(f"[emotion] start analyze filename={getattr(audio_file,'filename',None)}")| @test_router.post("/voice/analyze", response_model=VoiceAnalyzePreviewResponse) | ||
| async def test_emotion_analyze(file: UploadFile = File(...)): | ||
| try: | ||
| data = await file.read() | ||
| from io import BytesIO | ||
| class FileWrapper: | ||
| def __init__(self, content, filename): | ||
| self.file = content | ||
| self.filename = filename | ||
| self.content_type = "audio/m4a" if filename.lower().endswith(".m4a") else "audio/wav" | ||
| wrapped = FileWrapper(BytesIO(data), file.filename) | ||
| result = analyze_voice_emotion(wrapped) | ||
| probs = result.get("emotion_scores") or {} | ||
| def to_bps(x): | ||
| try: | ||
| return max(0, min(10000, int(round(float(x) * 10000)))) | ||
| except Exception: | ||
| return 0 | ||
| happy = to_bps(probs.get("happy", 0)) | ||
| sad = to_bps(probs.get("sad", 0)) | ||
| neutral = to_bps(probs.get("neutral", 0)) | ||
| angry = to_bps(probs.get("angry", 0)) | ||
| fear = to_bps(probs.get("fear", 0)) | ||
| surprise = to_bps(probs.get("surprise", 0)) | ||
| total = happy + sad + neutral + angry + fear + surprise | ||
| if total == 0: | ||
| neutral = 10000 | ||
| happy = sad = angry = fear = surprise = 0 | ||
| else: | ||
| scale = 10000 / float(total) | ||
| vals = { | ||
| "happy": int(round(happy * scale)), | ||
| "sad": int(round(sad * scale)), | ||
| "neutral": int(round(neutral * scale)), | ||
| "angry": int(round(angry * scale)), | ||
| "fear": int(round(fear * scale)), | ||
| "surprise": int(round(surprise * scale)), | ||
| } | ||
| diff = 10000 - sum(vals.values()) | ||
| if diff != 0: | ||
| k = max(vals, key=lambda k: vals[k]) | ||
| vals[k] = max(0, min(10000, vals[k] + diff)) | ||
| happy, sad, neutral, angry, fear, surprise = ( | ||
| vals["happy"], vals["sad"], vals["neutral"], vals["angry"], vals["fear"], vals["surprise"] | ||
| ) | ||
| top_emotion = result.get("top_emotion") or result.get("label") or result.get("emotion") | ||
| top_conf_bps = to_bps(result.get("top_confidence") or result.get("confidence", 0)) | ||
| return VoiceAnalyzePreviewResponse( | ||
| voice_id=None, | ||
| happy_bps=happy, | ||
| sad_bps=sad, | ||
| neutral_bps=neutral, | ||
| angry_bps=angry, | ||
| fear_bps=fear, | ||
| surprise_bps=surprise, | ||
| top_emotion=top_emotion, | ||
| top_confidence_bps=top_conf_bps, | ||
| model_version=result.get("model_version") | ||
| ) | ||
| except Exception as e: | ||
| raise HTTPException(status_code=400, detail=f"emotion analyze failed: {str(e)}") |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
정규화 로직이 voice_service와 중복됩니다.
Lines 289-320의 정규화 로직이 app/voice_service.py의 _process_audio_emotion_background (lines 170-223)와 거의 동일합니다. 공통 유틸리티 함수로 추출하여 중복을 제거하세요.
새 파일 app/emotion_utils.py를 생성:
def to_bps(v: float) -> int:
"""확률 값을 basis points (0-10000)로 변환"""
try:
return max(0, min(10000, int(round(float(v) * 10000))))
except Exception:
return 0
def normalize_emotion_scores(probs: dict) -> dict:
"""감정 점수를 정규화하여 합계가 10000이 되도록 함"""
happy = to_bps(probs.get("happy", probs.get("happiness", 0)))
sad = to_bps(probs.get("sad", probs.get("sadness", 0)))
neutral = to_bps(probs.get("neutral", 0))
angry = to_bps(probs.get("angry", probs.get("anger", 0)))
fear = to_bps(probs.get("fear", probs.get("fearful", 0)))
surprise = to_bps(probs.get("surprise", probs.get("surprised", 0)))
total_raw = happy + sad + neutral + angry + fear + surprise
if total_raw == 0:
return {"happy": 0, "sad": 0, "neutral": 10000, "angry": 0, "fear": 0, "surprise": 0}
scale = 10000 / float(total_raw)
vals = {
"happy": int(round(happy * scale)),
"sad": int(round(sad * scale)),
"neutral": int(round(neutral * scale)),
"angry": int(round(angry * scale)),
"fear": int(round(fear * scale)),
"surprise": int(round(surprise * scale)),
}
diff = 10000 - sum(vals.values())
if diff != 0:
key_max = max(vals, key=lambda k: vals[k])
vals[key_max] = max(0, min(10000, vals[key_max] + diff))
return vals그런 다음 voice_service.py와 main.py에서 이 함수를 import하여 사용하세요.
🤖 Prompt for AI Agents
In app/main.py around lines 276-336 the emotion normalization code duplicates
logic in app/voice_service.py (lines ~170-223); extract the shared helpers into
a new module app/emotion_utils.py with two functions: to_bps(v: float) -> int
that converts a probability to basis points safely, and
normalize_emotion_scores(probs: dict) -> dict that builds
happy/sad/neutral/angry/fear/surprise using sensible key fallbacks, scales them
to sum to 10000 (handling zero-total by returning neutral=10000), and fixes
rounding diff by adjusting the largest bucket; then replace the inline
normalization in both app/main.py and app/voice_service.py to import and call
normalize_emotion_scores(result.get("emotion_scores") or {}) and use to_bps for
top confidence conversion so both files use the shared utility.
| # 업로드 확장자에 맞춰 임시 파일로 저장 (기본: .wav) | ||
| orig_name = getattr(audio_file, "filename", "") or "" | ||
| _, ext = os.path.splitext(orig_name) | ||
| suffix = ext if ext.lower() in [".wav", ".m4a", ".mp3", ".flac", ".ogg", ".aac", ".caf"] else ".wav" | ||
|
|
||
| with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp_file: | ||
| content = audio_file.file.read() | ||
| audio_file.file.seek(0) | ||
| tmp_file.write(content) | ||
| tmp_file_path = tmp_file.name | ||
|
|
||
| # 오디오 파일 로드 및 전처리 | ||
| audio_data, sample_rate = librosa.load(tmp_file_path, sr=16000) | ||
| # 오디오 파일 로드 및 전처리 (견고한 로더) | ||
| def robust_load(path: str, target_sr: int = 16000): | ||
| """soundfile 우선, 실패 시 librosa로 폴백. 모노, 정규화 반환.""" | ||
| try: | ||
| data, sr = sf.read(path, always_2d=True, dtype="float32") # (N, C) | ||
| if data.ndim == 2 and data.shape[1] > 1: | ||
| data = data.mean(axis=1) # mono | ||
| else: | ||
| data = data.reshape(-1) | ||
| if sr != target_sr: | ||
| data = librosa.resample(data, orig_sr=sr, target_sr=target_sr) | ||
| sr = target_sr | ||
| return data, sr | ||
| except Exception: | ||
| # 폴백: librosa가 내부적으로 audioread/ffmpeg 사용 | ||
| y, sr = librosa.load(path, sr=target_sr, mono=True) | ||
| return y.astype("float32"), sr |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
오디오 로딩 로직 개선이 좋습니다만, 코드 중복을 제거하세요.
soundfile 우선, librosa 폴백 방식의 견고한 로딩 로직이 훌륭합니다. 하지만 app/emotion_service.py에 동일한 robust_load 함수가 중복되어 있습니다.
공통 유틸리티 모듈을 만들어 중복을 제거하세요:
새 파일 app/audio_utils.py 생성:
import librosa
import soundfile as sf
import numpy as np
def robust_load(path: str, target_sr: int = 16000):
"""soundfile 우선, 실패 시 librosa로 폴백. 모노, 정규화 반환."""
try:
data, sr = sf.read(path, always_2d=True, dtype="float32")
if data.ndim == 2 and data.shape[1] > 1:
data = data.mean(axis=1) # mono
else:
data = data.reshape(-1)
if sr != target_sr:
data = librosa.resample(data, orig_sr=sr, target_sr=target_sr)
sr = target_sr
return data, sr
except Exception:
y, sr = librosa.load(path, sr=target_sr, mono=True)
return y.astype("float32"), sr그런 다음 두 파일에서 import하여 사용:
from .audio_utils import robust_load| class TempUploadFile: | ||
| def __init__(self, content, filename): | ||
| self.file = content | ||
| self.filename = filename | ||
| self.content_type = "audio/m4a" if filename.endswith('.m4a') else "audio/wav" |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
TempUploadFile 클래스가 중복됩니다.
동일한 TempUploadFile 클래스가 _process_stt_and_nlp_background (lines 102-106)와 _process_audio_emotion_background (lines 151-155)에 중복 정의되어 있습니다. 클래스 레벨 또는 모듈 레벨 헬퍼로 추출하는 것을 권장합니다.
예시:
+class _TempUploadFile:
+ """음성 파일 처리를 위한 임시 래퍼"""
+ def __init__(self, content, filename):
+ self.file = content
+ self.filename = filename
+ self.content_type = "audio/m4a" if filename.endswith('.m4a') else "audio/wav"
+
class VoiceService:
"""음성 관련 서비스"""
def __init__(self, db: Session):
self.db = db
self.db_service = get_db_service(db)
self.auth_service = get_auth_service(db)그런 다음 두 메서드에서 _TempUploadFile을 재사용하세요.
Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In app/voice_service.py around lines 102-106 and 151-155, the TempUploadFile
class is duplicated inside two methods; extract it to module-level as a single
helper (e.g., name it _TempUploadFile) and replace the inner-class definitions
in both _process_stt_and_nlp_background and _process_audio_emotion_background
with references to that module-level class so both methods reuse the same
implementation; ensure the constructor signature and content_type logic ('.m4a'
-> "audio/m4a" else "audio/wav") remain identical and update any imports/uses
accordingly.
| def delete_user_voice(self, voice_id: int, username: str) -> Dict[str, Any]: | ||
| """사용자 소유 검증 후 음성 및 연관 데이터 삭제""" | ||
| try: | ||
| voice = self.db_service.get_voice_owned_by_username(voice_id, username) | ||
| if not voice: | ||
| return {"success": False, "message": "Voice not found or not owned by user"} | ||
|
|
||
| ok = self.db_service.delete_voice_with_relations(voice_id) | ||
| if not ok: | ||
| return {"success": False, "message": "Delete failed"} | ||
| return {"success": True, "message": "Deleted"} | ||
| except Exception as e: | ||
| return {"success": False, "message": f"Delete error: {str(e)}"} |
There was a problem hiding this comment.
예외 메시지가 내부 정보를 노출할 수 있습니다.
Line 373에서 str(e)를 클라이언트에 반환하면 내부 구현 세부사항이 노출될 수 있습니다. 프로덕션 환경에서는 일반적인 에러 메시지를 반환하고 자세한 내용은 로그에만 기록하는 것을 권장합니다.
def delete_user_voice(self, voice_id: int, username: str) -> Dict[str, Any]:
"""사용자 소유 검증 후 음성 및 연관 데이터 삭제"""
try:
voice = self.db_service.get_voice_owned_by_username(voice_id, username)
if not voice:
return {"success": False, "message": "Voice not found or not owned by user"}
ok = self.db_service.delete_voice_with_relations(voice_id)
if not ok:
return {"success": False, "message": "Delete failed"}
return {"success": True, "message": "Deleted"}
except Exception as e:
- return {"success": False, "message": f"Delete error: {str(e)}"}
+ print(f"Delete error for voice_id={voice_id}, username={username}: {e}", flush=True)
+ return {"success": False, "message": "Delete failed"}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def delete_user_voice(self, voice_id: int, username: str) -> Dict[str, Any]: | |
| """사용자 소유 검증 후 음성 및 연관 데이터 삭제""" | |
| try: | |
| voice = self.db_service.get_voice_owned_by_username(voice_id, username) | |
| if not voice: | |
| return {"success": False, "message": "Voice not found or not owned by user"} | |
| ok = self.db_service.delete_voice_with_relations(voice_id) | |
| if not ok: | |
| return {"success": False, "message": "Delete failed"} | |
| return {"success": True, "message": "Deleted"} | |
| except Exception as e: | |
| return {"success": False, "message": f"Delete error: {str(e)}"} | |
| def delete_user_voice(self, voice_id: int, username: str) -> Dict[str, Any]: | |
| """사용자 소유 검증 후 음성 및 연관 데이터 삭제""" | |
| try: | |
| voice = self.db_service.get_voice_owned_by_username(voice_id, username) | |
| if not voice: | |
| return {"success": False, "message": "Voice not found or not owned by user"} | |
| ok = self.db_service.delete_voice_with_relations(voice_id) | |
| if not ok: | |
| return {"success": False, "message": "Delete failed"} | |
| return {"success": True, "message": "Deleted"} | |
| except Exception as e: | |
| print(f"Delete error for voice_id={voice_id}, username={username}: {e}", flush=True) | |
| return {"success": False, "message": "Delete failed"} |
🤖 Prompt for AI Agents
In app/voice_service.py around lines 361 to 373, the exception handler returns
str(e) to the client which can leak internal details; instead, catch and log the
full exception (including stacktrace) to your application logger at error level,
then return a generic client-safe message like "Delete failed, please contact
support" or similar. Replace the direct str(e) return with process- or
module-level logger.error(...) capturing the exception, and return a
non-sensitive message and a suitable HTTP/internal error code as appropriate.
🔎 Description
🔗 Related Issue
🏷️ What type of PR is this?
📋 Changes Made
🧪 Testing
📸 Screenshots (if applicable)
📝 Additional Notes
Summary by CodeRabbit
릴리스 노트
새로운 기능
개선사항
문서